//
// Created by huyi on 24-7-12.
//
#include "DolphindbInstance.h"

#include "AmdUtil.h"
#include "QuoteAmaLiteService.h"

#include <folly/gen/Base.h>
#include <folly/gen/Core.h>

DECLARE_int32(queue_consumer_max_read_size);
DECLARE_string(ddb_username);
DECLARE_string(ddb_password);

namespace tradev::app {
using amd::ama::MarketType;

static constexpr const auto kBj  = ".BJ";
static constexpr const auto kCfe = ".CFE";
static constexpr const auto kSh  = ".SH";
static constexpr const auto kSz  = ".SZ";

/*!
 * @brief 从AMD代码转换为代码
 * @param code 代码, 如 000001
 * @param market 市场类型
 * @return 转换后的代码, 如 000001.SZ
 */
FOLLY_ALWAYS_INLINE
std::string normalizeAmdCode(folly::StringPiece code, uint8_t market) {
  using namespace folly;

  std::string rt;
  switch (market) {
  case MarketType::kNEEQ:
    toAppend(code, kBj, &rt);
    break;
  case MarketType::kCFFEX:
    toAppend(code, kCfe, &rt);
    break;
  case MarketType::kSSE:
    toAppend(code, kSh, &rt);
    break;
  case MarketType::kSZSE:
    toAppend(code, kSz, &rt);
    break;
  default:
    toAppend(code, ".0", &rt);
    break;
  }
  return std::move(rt);
}

int DolphindbInstance::open() {
  using namespace dolphindb;
  // 更新关闭标志
  _closed.store(false);

  CHECK_NOTNULL(_options["ddb-server"]);
  CHECK_NOTNULL(_options["ddb-port"]);

  auto server = _options["ddb-server"].asString();
  auto port   = _options["ddb-port"].asInt();

  XLOG(INFO) << "Connecting to the dolphin server: " << server << ":" << port;

  bool ret = _conn.connect(server, static_cast<int>(port));
  if (!ret) {
    XLOG(ERR, "Failed to connect to the dolphin server");
    return 1;
  }
  _conn.run(fmt::format("login('{}','{}');", FLAGS_ddb_username, FLAGS_ddb_password));
  _conn.run("use quote::Quote;");
  TableSP tb1 = _conn.run("loadStreamTableSnapshot()");
  XLOG(INFO) << "Table rows loaded: " << tb1->size();
  //
  _columns.clear();
  _column_names.clear();
  _column_types.clear();
  for (int i = 0; i < tb1->columns(); i++) {
    _columns.emplace_back(ColumnInfo{tb1->getColumnName(i), tb1->getColumnType(i)});
    _column_names.emplace_back(tb1->getColumnName(i));
    _column_types.emplace_back(tb1->getColumnType(i));
  }
  XLOG(INFO) << "Total columns: " << _columns.size();
  for (int i = 0; i < _columns.size(); i++) {
    XLOG(INFO) << "Column " << i << " name: " << _columns[i].name << " " << _columns[i].type;
  }

  return 0;
}

int DolphindbInstance::select1() {
  ConstantSP sp1 = _conn.run("1+0");
  if (sp1->isScalar()) {
    return sp1->getInt();
  } else {
    return 0;
  }
  return -1;
}

template<class T>
double MdSnapshotAppender::fixedClosePrice(const T* tick) {
  using namespace tradev::util;
  double close = 0;
  // YYYYMMDDHHMMSSsss
  auto time = util::extractTime(tick->orig_time);
  // 9点30分之前的收盘价需要特殊处理
  if (UNLIKELY(time < 93000000)) {
    if (LIKELY(tick->last_price > 0)) {
      close = convertPrice(tick->last_price);
    } else {
      // last_price有可能是0
      if (LIKELY(tick->bid_price[0] > 0 && tick->offer_price[0] > 0)) {
        // bid_price 和 ask_price均有值时取均值
        close = convertPrice((tick->offer_price[0] + tick->bid_price[0]) / 2);
      } else if (UNLIKELY(tick->bid_price[0] == 0 && tick->offer_price[0] != 0)) {
        // 跌停
        close = convertPrice(tick->offer_price[0]);
      } else if (UNLIKELY(tick->offer_price[0] == 0 && tick->bid_price[0] != 0)) {
        // 涨停
        close = convertPrice(tick->bid_price[0]);
      } else {
        // 可能停牌了,使用前收盘价
        close = convertPrice(tick->pre_close_price);
      }
    }
  } else { // 9点30分之后的收盘价正常处理
    close = convertPrice(
      // 收盘后才会有有效收盘价,否则使用最新价
      // 最新价无效时使用昨收价
      tick->close_price > 0
        ? tick->close_price
        : tick->last_price > 0
        ? tick->last_price
        : tick->pre_close_price);
  }
  return close;
}

template<>
void MdSnapshotAppender::fillSnapshotRow<MDSnapshot>(
  const TableSP&      table,
  int                 rowIdx,
  const MDSnapshot*   snapshot,
  const ConstantSP&   colDateTime,
  const ConstantSP&   colLastTime,
  DdbVector<double>&  bidPri,
  DdbVector<double>&  askPri,
  DdbVector<int64_t>& bidVol,
  DdbVector<int64_t>& askVol) {
  using namespace util;
  //
  auto code = normalizeAmdCode(snapshot->security_code, snapshot->market_type);

  table->getColumn(Fields::kDateTime)->set(rowIdx, colDateTime);
  table->getColumn(Fields::kCode)->setString(rowIdx, code);
  table->getColumn(Fields::kPhase)->setString(rowIdx, snapshot->trading_phase_code);
  table->getColumn(Fields::kPreClose)->setDouble(rowIdx, convertPrice(snapshot->pre_close_price));
  table->getColumn(Fields::kOpen)->setDouble(rowIdx, convertPrice(snapshot->open_price));
  table->getColumn(Fields::kHigh)->setDouble(rowIdx, convertPrice(snapshot->high_price));
  table->getColumn(Fields::kLow)->setDouble(rowIdx, convertPrice(snapshot->low_price));
  table->getColumn(Fields::kClose)->setDouble(rowIdx, fixedClosePrice(snapshot));
  table->getColumn(Fields::kLast)->setDouble(rowIdx, convertPrice(snapshot->last_price));
  table->getColumn(Fields::kBidPrice)->setDouble(rowIdx, convertPrice(snapshot->bid_price[0]));
  table->getColumn(Fields::kBidVolume)->setLong(rowIdx, convertVolume(snapshot->bid_volume[0]));
  table->getColumn(Fields::kAskPrice)->setDouble(rowIdx, convertPrice(snapshot->offer_price[0]));
  table->getColumn(Fields::kAskVolume)->setLong(rowIdx, convertVolume(snapshot->offer_volume[0]));
  table->getColumn(Fields::kBidPrices)->set(rowIdx, bidPri.createVector(dolphindb::DT_DOUBLE));
  table->getColumn(Fields::kBidVolumes)->set(rowIdx, bidVol.createVector(dolphindb::DT_LONG));
  table->getColumn(Fields::kAskPrices)->set(rowIdx, askPri.createVector(dolphindb::DT_DOUBLE));
  table->getColumn(Fields::kAskVolumes)->set(rowIdx, askVol.createVector(dolphindb::DT_LONG));
  table->getColumn(Fields::kNumTrades)->setLong(rowIdx, snapshot->num_trades);
  table->getColumn(Fields::kTotalVolumeTrade)->setLong(rowIdx, convertVolume(snapshot->total_volume_trade));
  table->getColumn(Fields::kTotalValueTrade)->setDouble(rowIdx, convertAmount(snapshot->total_value_trade));
  table->getColumn(Fields::kTotalBidVolume)->setLong(rowIdx, convertVolume(snapshot->total_bid_volume));
  table->getColumn(Fields::kTotalAskVolume)->setLong(rowIdx, convertVolume(snapshot->total_offer_volume));
  table->getColumn(Fields::kWeightedAvgBidPrice)->setDouble(rowIdx, convertPrice(snapshot->weighted_avg_bid_price));
  table->getColumn(Fields::kWeightedAvgAskPrice)->setDouble(rowIdx, convertPrice(snapshot->weighted_avg_offer_price));
  table->getColumn(Fields::kIOPV)->setDouble(rowIdx, convertPrice(snapshot->IOPV));
  table->getColumn(Fields::kYieldToMaturity)->setDouble(rowIdx, convertRate(snapshot->yield_to_maturity));
  table->getColumn(Fields::kHighLimited)->setDouble(rowIdx, convertPrice(snapshot->high_limited));
  table->getColumn(Fields::kLowLimited)->setDouble(rowIdx, convertPrice(snapshot->low_limited));
  table->getColumn(Fields::kStatus)->setString(rowIdx, snapshot->instrument_status);
  table->getColumn(Fields::kPreCloseIOPV)->setDouble(rowIdx, convertPrice(snapshot->pre_close_iopv));
  table->getColumn(Fields::kLastTradeTime)->set(rowIdx, colLastTime);
  table->getColumn(Fields::kVarietyCategory)->setString(rowIdx, folly::to<std::string>(snapshot->variety_category));
}

template<>
void MdSnapshotAppender::fillSnapshotRow<MDOptionSnapshot>(
  const TableSP&          table,
  int                     rowIdx,
  const MDOptionSnapshot* snapshot,
  const ConstantSP&       colDateTime,
  const ConstantSP&       colLastTime,
  DdbVector<double>&      bidPri,
  DdbVector<double>&      askPri,
  DdbVector<int64_t>&     bidVol,
  DdbVector<int64_t>&     askVol) {
  using namespace util;

  auto code = normalizeAmdCode(snapshot->security_code, snapshot->market_type);

  table->getColumn(Fields::kDateTime)->set(rowIdx, colDateTime);
  table->getColumn(Fields::kCode)->setString(rowIdx, code);
  table->getColumn(Fields::kPhase)->setString(rowIdx, snapshot->trading_phase_code);
  table->getColumn(Fields::kPreClose)->setDouble(rowIdx, convertPrice(snapshot->pre_close_price));
  table->getColumn(Fields::kOpen)->setDouble(rowIdx, convertPrice(snapshot->open_price));
  table->getColumn(Fields::kHigh)->setDouble(rowIdx, convertPrice(snapshot->high_price));
  table->getColumn(Fields::kLow)->setDouble(rowIdx, convertPrice(snapshot->low_price));
  table->getColumn(Fields::kClose)->setDouble(rowIdx, fixedClosePrice(snapshot));
  table->getColumn(Fields::kLast)->setDouble(rowIdx, convertPrice(snapshot->last_price));
  table->getColumn(Fields::kBidPrice)->setDouble(rowIdx, convertPrice(snapshot->bid_price[0]));
  table->getColumn(Fields::kBidVolume)->setLong(rowIdx, convertVolume(snapshot->bid_volume[0]));
  table->getColumn(Fields::kAskPrice)->setDouble(rowIdx, convertPrice(snapshot->offer_price[0]));
  table->getColumn(Fields::kAskVolume)->setLong(rowIdx, convertVolume(snapshot->offer_volume[0]));
  table->getColumn(Fields::kBidPrices)->set(rowIdx, bidPri.createVector(dolphindb::DT_DOUBLE));
  table->getColumn(Fields::kBidVolumes)->set(rowIdx, bidVol.createVector(dolphindb::DT_LONG));
  table->getColumn(Fields::kAskPrices)->set(rowIdx, askPri.createVector(dolphindb::DT_DOUBLE));
  table->getColumn(Fields::kAskVolumes)->set(rowIdx, askVol.createVector(dolphindb::DT_LONG));
  table->getColumn(Fields::kTotalVolumeTrade)->setLong(rowIdx, convertVolume(snapshot->total_volume_trade));
  table->getColumn(Fields::kTotalValueTrade)->setDouble(rowIdx, convertAmount(snapshot->total_value_trade));
  table->getColumn(Fields::kHighLimited)->setDouble(rowIdx, convertPrice(snapshot->high_limited));
  table->getColumn(Fields::kLowLimited)->setDouble(rowIdx, convertPrice(snapshot->low_limited));
  table->getColumn(Fields::kLastTradeTime)->set(rowIdx, colLastTime);
  table->getColumn(Fields::kVarietyCategory)->setString(rowIdx, folly::to<std::string>(snapshot->variety_category));
}

template<>
void MdSnapshotAppender::fillSnapshotRow<MDIndexSnapshot>(
  const TableSP&         table,
  int                    rowIdx,
  const MDIndexSnapshot* snapshot,
  const ConstantSP&      colDateTime,
  const ConstantSP&      colLastTime,
  DdbVector<double>&     bidPri,
  DdbVector<double>&     askPri,
  DdbVector<int64_t>&    bidVol,
  DdbVector<int64_t>&    askVol) {
  using namespace util;

  auto code = normalizeAmdCode(snapshot->security_code, snapshot->market_type);

  table->getColumn(Fields::kDateTime)->set(rowIdx, colDateTime);
  table->getColumn(Fields::kCode)->setString(rowIdx, code);
  table->getColumn(Fields::kPhase)->setString(rowIdx, snapshot->trading_phase_code);
  table->getColumn(Fields::kBidPrices)->set(rowIdx, bidPri.createVector(dolphindb::DT_DOUBLE));
  table->getColumn(Fields::kBidVolumes)->set(rowIdx, bidVol.createVector(dolphindb::DT_LONG));
  table->getColumn(Fields::kAskPrices)->set(rowIdx, askPri.createVector(dolphindb::DT_DOUBLE));
  table->getColumn(Fields::kAskVolumes)->set(rowIdx, askVol.createVector(dolphindb::DT_LONG));
  table->getColumn(Fields::kTotalVolumeTrade)->setLong(rowIdx, convertVolume(snapshot->total_volume_trade));
  table->getColumn(Fields::kTotalValueTrade)->setDouble(rowIdx, convertAmount(snapshot->total_value_trade));
  table->getColumn(Fields::kLastTradeTime)->set(rowIdx, colLastTime);
  table->getColumn(Fields::kVarietyCategory)->setString(rowIdx, folly::to<std::string>(snapshot->variety_category));
}


template<>
void MdSnapshotAppender::fillSnapshotRow<MDFutureSnapshot>(
  const TableSP&          table,
  int                     rowIdx,
  const MDFutureSnapshot* snapshot,
  const ConstantSP&       colDateTime,
  const ConstantSP&       colLastTime,
  DdbVector<double>&      bidPri,
  DdbVector<double>&      askPri,
  DdbVector<int64_t>&     bidVol,
  DdbVector<int64_t>&     askVol) {
  using namespace util;
  //
  auto code = normalizeAmdCode(snapshot->security_code, snapshot->market_type);

  table->getColumn(Fields::kDateTime)->set(rowIdx, colDateTime);
  table->getColumn(Fields::kCode)->setString(rowIdx, code);
  table->getColumn(Fields::kPreClose)->setDouble(rowIdx, convertPrice(snapshot->pre_close_price));
  table->getColumn(Fields::kOpen)->setDouble(rowIdx, convertPrice(snapshot->open_price));
  table->getColumn(Fields::kHigh)->setDouble(rowIdx, convertPrice(snapshot->high_price));
  table->getColumn(Fields::kLow)->setDouble(rowIdx, convertPrice(snapshot->low_price));
  table->getColumn(Fields::kClose)->setDouble(rowIdx, fixedClosePrice(snapshot));
  table->getColumn(Fields::kLast)->setDouble(rowIdx, convertPrice(snapshot->last_price));
  table->getColumn(Fields::kBidPrice)->setDouble(rowIdx, convertPrice(snapshot->bid_price[0]));
  table->getColumn(Fields::kBidVolume)->setLong(rowIdx, convertVolume(snapshot->bid_volume[0]));
  table->getColumn(Fields::kAskPrice)->setDouble(rowIdx, convertPrice(snapshot->offer_price[0]));
  table->getColumn(Fields::kAskVolume)->setLong(rowIdx, convertVolume(snapshot->offer_volume[0]));
  table->getColumn(Fields::kBidPrices)->set(rowIdx, bidPri.createVector(dolphindb::DT_DOUBLE));
  table->getColumn(Fields::kBidVolumes)->set(rowIdx, bidVol.createVector(dolphindb::DT_LONG));
  table->getColumn(Fields::kAskPrices)->set(rowIdx, askPri.createVector(dolphindb::DT_DOUBLE));
  table->getColumn(Fields::kAskVolumes)->set(rowIdx, askVol.createVector(dolphindb::DT_LONG));
  table->getColumn(Fields::kTotalVolumeTrade)->setLong(rowIdx, convertVolume(snapshot->total_volume_trade));
  table->getColumn(Fields::kTotalValueTrade)->setDouble(rowIdx, convertAmount(snapshot->total_value_trade));
  table->getColumn(Fields::kHighLimited)->setDouble(rowIdx, convertPrice(snapshot->high_limited));
  table->getColumn(Fields::kLowLimited)->setDouble(rowIdx, convertPrice(snapshot->low_limited));
  table->getColumn(Fields::kLastTradeTime)->set(rowIdx, colLastTime);
  table->getColumn(Fields::kVarietyCategory)->setString(rowIdx, folly::to<std::string>(snapshot->variety_category));
}

template<class T>
void MdSnapshotAppender::fillSnapshot(T* p, int rowIdx, dolphindb::TableSP& out) {
  using namespace util;

  auto origTime = parseDateTime(folly::to<folly::fbstring>(p->orig_time));
  auto lastTime = parseDateTime(folly::to<folly::fbstring>(p->last_trade_time));

  ConstantSP colDateTime = dolphindb::Util::createTimestamp(origTime[0], origTime[1], origTime[2], origTime[3], origTime[4], origTime[5], origTime[6]);
  ConstantSP colLastTime = dolphindb::Util::createTimestamp(lastTime[0], lastTime[1], lastTime[2], lastTime[3], lastTime[4], lastTime[5], lastTime[6]);

  DdbVector<double>  bp1(5, 5);
  DdbVector<double>  ap1(5, 5);
  DdbVector<int64_t> bv1(5, 5);
  DdbVector<int64_t> av1(5, 5);
  for (int depth = 0; depth < 5; ++depth) {
    bp1.set(depth, convertPrice(p->bid_price[depth]));
    ap1.set(depth, convertPrice(p->offer_price[depth]));
    bv1.set(depth, convertVolume(p->bid_volume[depth]));
    av1.set(depth, convertVolume(p->offer_volume[depth]));
  }
  fillSnapshotRow(out, rowIdx, p, colDateTime, colLastTime, bp1, ap1, bv1, av1);
}

/*!
 * @brief 特化指数版本
 * @param p
 * @param rowIdx
 * @param out
 */
template<>
void MdSnapshotAppender::fillSnapshot(MDIndexSnapshot* p, int rowIdx, dolphindb::TableSP& out) {
  using namespace util;
  using namespace dolphindb;
  auto origTime = parseDateTime(folly::to<folly::fbstring>(p->orig_time));
  auto lastTime = parseDateTime(folly::to<folly::fbstring>(p->orig_time));

  ConstantSP colDateTime = Util::createTimestamp(origTime[0], origTime[1], origTime[2], origTime[3], origTime[4], origTime[5], origTime[6]);
  ConstantSP colLastTime = Util::createTimestamp(lastTime[0], lastTime[1], lastTime[2], lastTime[3], lastTime[4], lastTime[5], lastTime[6]);

  DdbVector<double>  bp1(5, 5);
  DdbVector<double>  ap1(5, 5);
  DdbVector<int64_t> bv1(5, 5);
  DdbVector<int64_t> av1(5, 5);
  fillSnapshotRow(out, rowIdx, p, colDateTime, colLastTime, bp1, ap1, bv1, av1);
}

/*!
 * @brief 特化期货版本
 */
template<>
void MdSnapshotAppender::fillSnapshot(MDFutureSnapshot* p, int rowIdx, dolphindb::TableSP& out) {
  using namespace dolphindb;
  using namespace util;

  auto origTime = parseDateTime(folly::to<folly::fbstring>(p->orig_time));
  auto lastTime = parseDateTime(folly::to<folly::fbstring>(p->orig_time));

  ConstantSP colDateTime = Util::createTimestamp(origTime[0], origTime[1], origTime[2], origTime[3], origTime[4], origTime[5], origTime[6]);
  ConstantSP colLastTime = Util::createTimestamp(lastTime[0], lastTime[1], lastTime[2], lastTime[3], lastTime[4], lastTime[5], lastTime[6]);

  DdbVector<double>  bp1(5, 5);
  DdbVector<double>  ap1(5, 5);
  DdbVector<int64_t> bv1(5, 5);
  DdbVector<int64_t> av1(5, 5);
  for (int depth = 0; depth < 5; ++depth) {
    bp1.set(depth, convertPrice(p->bid_price[depth]));
    ap1.set(depth, convertPrice(p->offer_price[depth]));
    bv1.set(depth, convertVolume(p->bid_volume[depth]));
    av1.set(depth, convertVolume(p->offer_volume[depth]));
  }
  fillSnapshotRow(out, rowIdx, p, colDateTime, colLastTime, bp1, ap1, bv1, av1);
}

/*!
 * @brief
 * @param idx
 * @param pItem
 */
size_t MdSnapshotAppender::append(ItemPtr&& pItem) {
  //    XLOG(INFO) << pItem->md->security_code;
  _item_cache.emplace_back(std::move(pItem));
  return _item_cache.size();
}

/*!
 * @brief
 */
void MdSnapshotAppender::flush() {
  using namespace dolphindb;

  // 处理异常
  if (!_service->checkDolphindbHealth()) {
    flushFailed();
    return;
  }

  static std::vector<std::string> colNames = _service->dolphindb()->column_names();
  static std::vector<DATA_TYPE>   colTypes = _service->dolphindb()->column_types();

  // 每次插入的最大行数
  static const auto kMaxRowSize = FLAGS_queue_consumer_max_read_size;

  // 使用函数内的静态局部变量初始化列向量，只初始化一次
  //  static std::vector<ConstantSP> colVec = [] {
  //    std::vector<ConstantSP> vec;
  //    vec.reserve(colTypes.size());
  //    for (const auto& colType : colTypes) {
  //      vec.emplace_back(Util::createVector(colType, kMaxRowSize, kMaxRowSize));
  //    }
  //    return vec;
  //  }();

  // 没有元素
  if (_item_cache.empty()) {
    return;
  }

  // 统计总行数
  auto rowCnt = folly::to<int>(countCachedRows());
  CHECK_LE(rowCnt, FLAGS_queue_consumer_max_read_size);
  CHECK_GE(rowCnt, _item_cache.size());

  // 创建待插入表格
  TableSP table = Util::createTable(colNames, colTypes, rowCnt, rowCnt);

  // 赋值
  int rowIdx = 0;
  for (const auto& e : _item_cache) {
    for (int ptrOffset = 0; ptrOffset < e->size; ptrOffset++) {
      if (e->type == kMdSnapshot) {
        fillSnapshot(e->stocks + ptrOffset, rowIdx, table);
      } else if (e->type == kMdOptionSnapshot) {
        fillSnapshot(e->options + ptrOffset, rowIdx, table);
      } else if (e->type == kMdIndexSnapshot) {
        fillSnapshot(e->indexes + ptrOffset, rowIdx, table);
      } else if (e->type == kMdFutureSnapshot) {
        fillSnapshot(e->futures + ptrOffset, rowIdx, table);
      }

      // 当前行处理完成
      ++rowIdx;
    }
  }

  std::vector<ConstantSP> args{table};

  auto rc = _service->dolphindb()->run("tableInsert{objByName(`snapshot_default_s)}", args)->getInt();
  CHECK_EQ(rc, rowCnt) << "插入数量不符";

  // 缓存重置
  _item_cache.clear();

  //
  XLOGF_IF(INFO, rowCnt >= 100, "Write {} items", rowCnt);
}

void MdSnapshotAppender::flushFailed() {
  auto size = _item_cache.size();
  // 使用 folly::gen 计算总和
  auto totalSize = folly::gen::from(_item_cache)
    | folly::gen::map([](const auto& e) { return e->size; })
    | folly::gen::sum;
  _item_cache.clear();
  XLOG(WARN) << "Drop " << totalSize << " items";
}

} // namespace tradev::app
